/*
 * Copyright (c) 2022. China Mobile (SuZhou) Software Technology Co.,Ltd. All rights reserved.
 * Lakehouse is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

package com.chinamobile.cmss.lakehouse.core.deploy.cluster.instance;

import com.chinamobile.cmss.lakehouse.common.dto.LakehouseCreateRequest;
import com.chinamobile.cmss.lakehouse.common.enums.ComponentTypeEnum;
import com.chinamobile.cmss.lakehouse.core.config.CommonOptions;
import com.chinamobile.cmss.lakehouse.core.config.ComponentConfiguration;
import com.chinamobile.cmss.lakehouse.core.config.KubernetesConfiguration;
import com.chinamobile.cmss.lakehouse.core.config.S3Config;
import com.chinamobile.cmss.lakehouse.core.deploy.cluster.descriptor.ClusterComponentSpec;
import com.chinamobile.cmss.lakehouse.core.deploy.cluster.descriptor.ClusterDescriptor;
import com.chinamobile.cmss.lakehouse.core.deploy.component.instance.hive.HiveComponentDescriptor;
import com.chinamobile.cmss.lakehouse.core.deploy.component.instance.hive.HiveComponentOptions;
import com.chinamobile.cmss.lakehouse.core.deploy.component.instance.meta.MetaComponentDescriptor;
import com.chinamobile.cmss.lakehouse.core.deploy.component.instance.meta.MetaComponentOptions;
import com.chinamobile.cmss.lakehouse.core.deploy.component.instance.yunikorn.YuniKornComponentDescriptor;
import com.chinamobile.cmss.lakehouse.core.deploy.component.instance.yunikorn.YuniKornComponentOptions;

import java.util.List;

import javax.annotation.Resource;

import com.google.common.collect.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.stereotype.Component;

@Component
public class LakehouseClusterDescriptor extends ClusterDescriptor {

    @Resource
    private HiveComponentDescriptor hiveDescriptor;

    @Resource
    private MetaComponentDescriptor metaDescriptor;

    @Resource
    private YuniKornComponentDescriptor yuniKornDescriptor;

    @Autowired
    private DataSourceProperties dataSourceProperties;

    @Autowired
    private S3Config s3Config;

    @Override
    public List<ClusterComponentSpec> clusterComponentSpecs() {
        List<ClusterComponentSpec> clusterComponentSpecList = Lists.newArrayList();
        clusterComponentSpecList.add(new ClusterComponentSpec(ComponentTypeEnum.HIVE, true));
        clusterComponentSpecList.add(new ClusterComponentSpec(ComponentTypeEnum.META, true));
        clusterComponentSpecList.add(new ClusterComponentSpec(ComponentTypeEnum.YUNIKORN, true));

        return clusterComponentSpecList;
    }

    /**
     * set component basic config
     *
     * @param request
     * @return
     */
    @Override
    public ComponentConfiguration buildComponentConfig(LakehouseCreateRequest request) {
        // set basic config
        ComponentConfiguration config = buildBasicConfig(request);

        return config;
    }

    public ComponentConfiguration buildBasicConfig(LakehouseCreateRequest request) {
        ComponentConfiguration config = new ComponentConfiguration();

        // set resource size
        config.set(CommonOptions.RESOURCE_SIZE, request.getResouceSize().getType());
        // set template replicas
        config.set(HiveComponentOptions.HIVE_REPLICA, KubernetesConfiguration.hiveReplicas);
        config.set(MetaComponentOptions.META_REPLICA, KubernetesConfiguration.metaReplicas);
        config.set(YuniKornComponentOptions.YUNIKORN_REPLICA, KubernetesConfiguration.yunikornReplicas);

        // set meta mysql info
        config.set(MetaComponentOptions.META_MYSQL_ADDRESS, getMetadataUrl());
        config.set(MetaComponentOptions.META_MYSQL_USER, dataSourceProperties.getUsername());
        config.set(MetaComponentOptions.META_MYSQL_PASS, dataSourceProperties.getPassword());

        // set meta s3 info
        config.set(MetaComponentOptions.META_DEFAULT_FS, getDefaultFsUri());
        config.set(MetaComponentOptions.META_WAREHOUSE_DIR, getWarehouseDir());
        config.set(MetaComponentOptions.META_S3_ENDPOINT, s3Config.getEndpoint());
        config.set(MetaComponentOptions.META_S3_ACCESS_KEY, s3Config.getAccessKey());
        config.set(MetaComponentOptions.META_S3_SECRET_KEY, s3Config.getSecretKey());

        // set hive s3 info
        config.set(HiveComponentOptions.HIVE_DEFAULT_FS, getDefaultFsUri());
        config.set(HiveComponentOptions.HIVE_WAREHOUSE_DIR, getWarehouseDir());
        config.set(HiveComponentOptions.HIVE_S3_ENDPOINT, s3Config.getEndpoint());
        config.set(HiveComponentOptions.HIVE_S3_ACCESS_KEY, s3Config.getAccessKey());
        config.set(HiveComponentOptions.HIVE_S3_SECRET_KEY, s3Config.getSecretKey());

        return config;
    }

    private String getMetadataUrl() {
        String url = dataSourceProperties.getUrl();
        String hiveSchemaDbSuffix = "_hive_schema";
        return url.substring(0, url.lastIndexOf("?")) + hiveSchemaDbSuffix + url.substring(url.lastIndexOf("?"));
    }

    private String getDefaultFsUri() {
        return "s3a://" + s3Config.getBucket();
    }

    private String getWarehouseDir() {
        String bucket = s3Config.getBucket();
        String dir = bucket.endsWith("/") ? bucket : bucket + "/";
        return "s3a://" + dir;
    }

    /**
     * check common component: hive, meta
     *
     * @return
     */
    @Override
    public boolean isCommonRunning() {
        boolean hiveRunning = hiveDescriptor.isRunning(commonNamespace());
        boolean metaRunning = metaDescriptor.isRunning(commonNamespace());
        boolean yuniKornRunning = yuniKornDescriptor.isRunning(commonNamespace());

        return hiveRunning && metaRunning && yuniKornRunning;
    }

    @Override
    public boolean isCommonResourceExisting() {
        boolean hiveExisting = hiveDescriptor.isCommonResourceExisting(commonNamespace());
        boolean metaExisting = metaDescriptor.isCommonResourceExisting(commonNamespace());
        boolean yuniKornExisting = yuniKornDescriptor.isCommonResourceExisting(commonNamespace());

        return hiveExisting && metaExisting && yuniKornExisting;
    }

    @Override
    public boolean isRunning(String clusterName) {
        return isCommonRunning();
    }
}
